package com.networkflow_analysis;

import com.networkflow_analysis.bean.PageViewCount;
import com.networkflow_analysis.bean.UserBehavior;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/10 23:24
 * <p>
 * 网站独立访客数（UV）的统计
 */
public class UniqueVisitor {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        String filePath = "NetworkFlowAnalysis/src/main/resources/UserBehavior.csv";
        DataStream<String> fileStream = env.readTextFile(filePath);

        //转换为POJO，分配时间戳和watermark
        DataStream<UserBehavior> dataStream = fileStream
                .map(line -> {
                    String[] fields = line.split(",");
                    return new UserBehavior(new Long(fields[0]), new Long(fields[1]), new Integer(fields[2]), fields[3], new Long(fields[4]));
                })
                /**
                 * 有序窗口
                 */
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
                    @Override
                    public long extractAscendingTimestamp(UserBehavior userBehavior) {
                        return userBehavior.getTimestamp() * 1000L;
                    }
                });
        /**
         * 开窗统计uv值
         */
        DataStream<PageViewCount> uvStream = dataStream
                .filter(data -> "pv".equals(data.getBehavior()))
                .timeWindowAll(Time.hours(1))
                .apply(new UvCountResult());

        uvStream.print();

        env.execute("uv count job");
    }

    /**
     * 实现自定义全窗口函数
     */
    public static class UvCountResult implements AllWindowFunction<UserBehavior, PageViewCount, TimeWindow>{
        @Override
        public void apply(TimeWindow timeWindow, Iterable<UserBehavior> iterable, Collector<PageViewCount> collector) throws Exception {
            //定义一个Set结构，保存窗口中的所有userId,自动去重
            HashSet<Long> uidSet = new HashSet<>();
            for(UserBehavior ub:iterable)
                uidSet.add(ub.getUserId());
            collector.collect(new PageViewCount("uv",timeWindow.getEnd(),new Long(uidSet.size())));

        }
    }
}
